package avicit.bdp.dcs.job.service;

import avicit.bdp.common.base.BaseService;
import avicit.bdp.common.datasource.ApiDataSource;
import avicit.bdp.common.datasource.BaseDataSource;
import avicit.bdp.common.datasource.DataSourceFactory;
import avicit.bdp.common.dto.ProcessInstanceDto;
import avicit.bdp.common.dto.TaskInstanceDto;
import avicit.bdp.common.dto.api.APIHttpAuth;
import avicit.bdp.common.dto.api.APIHttpDefinition;
import avicit.bdp.common.dto.query.dds.ExecutorQuery;
import avicit.bdp.common.dto.request.dds.ProcessSchedulerDto;
import avicit.bdp.common.dto.request.dds.ScheduleParamDto;
import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.service.dto.TaskSpeedDTO;
import avicit.bdp.common.service.service.DataSourceService;
import avicit.bdp.common.utils.BdpLogUtil;
import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.common.utils.DateUtils;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.common.utils.uploads.CatalogBean;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.file.util.FileUtils;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.alert.dto.AlertDTO;
import avicit.bdp.dcs.alert.service.AlertService;
import avicit.bdp.dcs.datasource.service.BaseDbService;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.datasource.service.TableService;
import avicit.bdp.dcs.detail.dto.TableDetailDto;
import avicit.bdp.dcs.detail.dto.TaskDataDetailDTO;
import avicit.bdp.dcs.detail.dto.TaskTableDetailDTO;
import avicit.bdp.dcs.detail.service.TaskDataDetailService;
import avicit.bdp.dcs.detail.service.TaskTableDetailService;
import avicit.bdp.dcs.job.dao.DcsJobDao;
import avicit.bdp.dcs.job.dto.*;
import avicit.bdp.dcs.json.service.DataxJsonService;
import avicit.bdp.dcs.json.service.TaskJsonService;
import avicit.bdp.dcs.log.dto.JobLogDTO;
import avicit.bdp.dcs.log.dto.ProcessDefinitionDto;
import avicit.bdp.dcs.log.service.DcsTaskLogService;
import avicit.bdp.dcs.precheck.dto.DcsJobPrecheckDto;
import avicit.bdp.dcs.precheck.dto.JobPrecheckResultDTO;
import avicit.bdp.dcs.precheck.service.JobPrecheckResultService;
import avicit.bdp.dcs.rule.dto.ColumnRuleDTO;
import avicit.bdp.dcs.rule.dto.TableRuleDTO;
import avicit.bdp.dcs.rule.service.ColumnRuleService;
import avicit.bdp.dcs.rule.service.TableRuleService;
import avicit.bdp.dcs.semistructure.dto.*;
import avicit.bdp.dcs.semistructure.service.FileTemplateService;
import avicit.bdp.dcs.semistructure.service.JobFileService;
import avicit.bdp.dcs.stream.dto.KafkaJsonDto;
import avicit.bdp.dcs.stream.dto.MsgTemplateDto;
import avicit.bdp.dcs.stream.service.MsgTemplateService;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.StreamColumnMappingDto;
import avicit.bdp.dcs.task.dto.TableMappingDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.task.service.ApiDefinitionService;
import avicit.bdp.dcs.task.service.StructureTaskService;
import avicit.bdp.dcs.task.service.TaskService;
import avicit.bdp.dcs.tools.enums.*;
import avicit.bdp.dcs.utils.DcsConstants;
import avicit.bdp.dcs.utils.DcsErrorCode;
import avicit.bdp.dcs.utils.PlatformUtil;
import avicit.bdp.dcs.utils.api.Configuration;
import avicit.bdp.dcs.utils.api.HttpUtil;
import avicit.bdp.dds.api.ExecutorApi;
import avicit.bdp.dds.api.ProcessDefinitionApi;
import avicit.bdp.dds.api.SchedulerApi;
import avicit.bdp.dds.dispatch.enums.CommandType;
import avicit.bdp.dds.dispatch.enums.ExecutionStatus;
import avicit.bdp.dds.dispatch.process.Property;
import avicit.platform6.api.system.ConvertColumnClient;
import avicit.platform6.commons.utils.BusinessUtil;
import avicit.platform6.commons.utils.ComUtil;
import avicit.platform6.core.context.ThreadContextHelper;
import avicit.platform6.core.exception.BusinessException;
import avicit.platform6.core.properties.PlatformConstant;
import avicit.platform6.core.rest.msg.QueryRespBean;
import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.TypeReference;
import com.github.pagehelper.Page;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.text.DecimalFormat;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @金航数码科技有限责任公司 @作者：liyb @邮箱：liyb@avic-digital.com @创建时间： 2020-11-10 18:53 @类说明： 作业服务类 @修改记录：
 */
@Service
public class DcsJobService extends BaseService<DcsJobDao, JobDTO> {

  private static final Logger logger = LoggerFactory.getLogger(DcsJobService.class);

  protected static Map<String, String> jobProcessInstanceMap = new HashMap<>();

  @Autowired private DcsJobDao jobDao;
  @Autowired private JobPrecheckResultService jobPrecheckResultService;
  @Autowired private TableRuleService tableRuleService;
  @Autowired private ColumnRuleService columnRuleService;
  @Autowired private TaskTableDetailService taskTableDetailService;
  @Autowired private TaskDataDetailService taskDataDetailService;
  @Autowired private TaskJsonService taskJsonService;
  @Autowired private TaskService taskService;
  @Autowired private AlertService alertService;
  @Autowired private DcsTaskLogService taskLogService;
  @Autowired private DataxJsonService dataxJsonService;
  @Autowired private ExecutorApi executorApi;
  @Autowired private SchedulerApi schedulerApi;
  @Autowired private ProcessDefinitionApi processDefinitionApi;
  @Autowired private JobFileService jobFileService;
  @Autowired private DataSourceService dataSourceService;
  @Autowired private MsgTemplateService msgTemplateService;
  @Autowired private JobFactory jobFactory;
  @Autowired private DataSourceUtils dataSourceUtils;
  @Autowired private BaseDbService baseDbService;
  @Autowired private ApiDefinitionService apiDefinitionService;
  @Autowired private FileTemplateService fileTemplateService;
  @Autowired private TableService tableService;
  @Autowired private StructureTaskService structureTaskService;
  @Autowired private ConvertColumnClient convertColumnClient;

  // 创建线程池
  private ThreadPoolExecutor threadPool =
      new ThreadPoolExecutor(
          5,
          10,
          1000L,
          TimeUnit.MILLISECONDS,
          new LinkedBlockingQueue<Runnable>(1024),
          new ThreadFactory() {
            public Thread newThread(Runnable r) {
              // 线程命名
              return new Thread(r, "threadPool" + r.hashCode());
            }
          },
          new ThreadPoolExecutor.CallerRunsPolicy());

  // 为了避免调用BaseService的更新方法，把修改时间写入
  @Transactional
  public void updateByPrimaryKeySelective(JobDTO jobDTO) {
    jobDao.updateByPrimaryKeySelective(jobDTO);
  }

  /** 查询作业列表 */
  public QueryRespBean<JobDTO> getJobList(
      Integer jobStatus,
      Integer sort,
      String name,
      Integer jobType,
      String projectId,
      Integer currentPage,
      Integer pageSize)
      throws Exception {
    long startTime = System.currentTimeMillis();

    // 排序，按修改时间倒序：1，按修改时间升序：2，按job名称升序：3，按job名称倒序：4
    String orderField = DcsConstants.LAST_UPDATE_DATE;
    String orderType = DcsConstants.DESC;
    if (sort != null) {
      if (1 == sort) {
        orderField = DcsConstants.LAST_UPDATE_DATE;
        orderType = DcsConstants.DESC;
      } else if (2 == sort) {
        orderField = DcsConstants.LAST_UPDATE_DATE;
        orderType = DcsConstants.ASC;
      } else if (3 == sort) {
        orderField = DcsConstants.NAME;
        orderType = DcsConstants.ASC;
      } else if (4 == sort) {
        orderField = DcsConstants.NAME;
        orderType = DcsConstants.DESC;
      } else {
        throw new BusinessException("不支持的排序类型");
      }
    }

    // 文档密级
    List<String> wordSecretList = ThreadContextHelper.getWordSecretList();
    QueryRespBean<JobDTO> queryRespBean = new QueryRespBean<>();
    com.github.pagehelper.PageHelper.startPage(currentPage, pageSize);

    Page<JobDTO> jobList =
        jobDao.getJobList(jobStatus, name, jobType, orderField, orderType, null, projectId, wordSecretList);
    if (CollectionUtils.isNotEmpty(jobList)) {
      List<ProcessInstanceDto> processInstanceList =
          jobDao.getProcessInstanceListByProcessDefinitionId(jobList);
      Map<String, ProcessInstanceDto> processInstanceMap = new HashMap<>();
      if (CollectionUtils.isNotEmpty(processInstanceList)) {
        processInstanceMap =
            processInstanceList.stream()
                .collect(
                    Collectors.toMap(
                        ProcessInstanceDto::getProcessDefinitionId, Function.identity()));
      }

      for (JobDTO jobDTO : jobList) {
        // 结合流程实例得出job状态
        this.handleStatus(jobDTO, processInstanceMap);

        // 结构化处理ExtraParam
        handleExtraParam(jobDTO);

        // 流式处理KafkaJson
        handleKafkaJson(jobDTO);

        // 设置目标库类型
        setDstDbType(jobDTO);

        // API采集获取API定义
        getAPIHttpDefinition(jobDTO);
      }
    }

    valueConvert(jobList);
    queryRespBean.setResult(jobList);
    logger.info("查询耗时:{}ms", System.currentTimeMillis() - startTime);
    BdpLogUtil.log4Query(queryRespBean);
    return queryRespBean;
  }

  private void valueConvert(List<JobDTO> jobList) {
    // 循环组装请求数据
    Map<String, Set<String>> convertFormData = new HashMap<>();
    if (CollectionUtils.isNotEmpty(jobList)) {
      for (JobDTO jobDTO : jobList) {
        BusinessUtil.createConvertSet(convertFormData, Constants.BDP_DATA_SECRET_LEVEL_KEY, jobDTO.getSecretLevel());
      }
      if (convertFormData.size() > 0) {
        // 获取请求结果
        Map<String, Map<String, String>> convertResultData = convertColumnClient.replace(convertFormData);
        for (JobDTO jobDTO : jobList) {
          jobDTO.setSecretLevelName(BusinessUtil.convertFormat(
                  convertResultData, Constants.BDP_DATA_SECRET_LEVEL_KEY, jobDTO.getSecretLevel()));
        }
      }
    }
  }

  /**
   * 查看作业详情
   *
   * @param jobId
   * @return
   */
  public JobDTO getJobById(String jobId) {
    JobDTO jobDTO = this.selectByPrimaryKey(jobId);
    // 结构化处理ExtraParam
    handleExtraParam(jobDTO);

    // 流式处理KafkaJson
    handleKafkaJson(jobDTO);

    // 设置目标库类型
    setDstDbType(jobDTO);

    // API采集获取API定义
    getAPIHttpDefinition(jobDTO);
    return jobDTO;
  }

  /** 结构化处理ExtraParam */
  private void handleExtraParam(JobDTO jobDTO) {
    if (jobDTO.getJobType() == JobType.TRANSFER.getCode()
        || jobDTO.getJobType() == JobType.SYNC.getCode()) {
      if (StringUtils.isNotEmpty(jobDTO.getExtraParams())) {
        HiveJsonDto hiveJsonDto =
            JSONObject.parseObject(jobDTO.getExtraParams(), HiveJsonDto.class);
        jobDTO.setHiveJsonDto(hiveJsonDto);
      }
    }
  }

  /**
   * 流式处理KafkaJson
   *
   * @param jobDTO
   */
  private void handleKafkaJson(JobDTO jobDTO) {
    if (StringUtils.isNotEmpty(jobDTO.getKafkaJson())) {
      KafkaJsonDto kafkaJsonDto = JSONObject.parseObject(jobDTO.getKafkaJson(), KafkaJsonDto.class);
      jobDTO.setKafkaJsonDto(kafkaJsonDto);
    }
  }

  /**
   * 设置目标库类型
   *
   * @param jobDTO
   */
  private void setDstDbType(JobDTO jobDTO) {
    // 半结构化、流式用于控制显示iotdb设备名或者hive表名
    if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()
        || jobDTO.getJobType() == JobType.STREAM.getCode()) {
      try {
        DataSourceDTO dataSourceDTO = dataSourceUtils.getDatasource(jobDTO.getDstDatasourceId());
        jobDTO.setDstDbType(dataSourceDTO.getDatasourceType());
      } catch (Exception e) {
        logger.error("查询数据源出错：{}", e.getMessage());
      }
    }
  }

  /**
   * API采集获取API定义
   *
   * @param jobDTO
   */
  private void getAPIHttpDefinition(JobDTO jobDTO) {
    if (jobDTO.getJobType() == JobType.API_HTTP.getCode()) {
      try {
        APIHttpDefinition apiHttpDefinition =
            apiDefinitionService.getAPIHttpDefinition(jobDTO.getId());
        jobDTO.setApiHttpDefinition(apiHttpDefinition);
      } catch (Exception e) {
        logger.error(e.getMessage());
        jobDTO.setApiHttpDefinition(apiDefinitionService.getAPIHttpDefinitionInstance());
      }
    }
  }

  /**
   * 结合流程实例得出job状态
   *
   * @param jobDTO job dto
   */
  private void handleStatus(JobDTO jobDTO, Map<String, ProcessInstanceDto> processInstanceMap) {
    ProcessInstanceDto processInstanceDto = processInstanceMap.get(jobDTO.getId());
    String processInstanceId = jobProcessInstanceMap.get(jobDTO.getId());

    // 可能会出现调度系统还未生成最新的流程实例，但是采集系统调用了获取列表服务，导致当前获取的流程实例不是最新的
    if (processInstanceDto != null && !processInstanceDto.getId().equals(processInstanceId)) {

      if (jobDTO.getTaskAvgSpeed() != null
          && !isEditStatus(jobDTO)
          && processInstanceDto.getId().equals(jobDTO.getProcessInstanceId())) {
        jobDTO.setTaskAvgSpeedUnit(FileUtils.readableSpeedSize(jobDTO.getTaskAvgSpeed()));
      }
      if (jobDTO.getReadSucceedBytes() != null
          && !isEditStatus(jobDTO)
          && processInstanceDto.getId().equals(jobDTO.getProcessInstanceId())) {
        jobDTO.setReadSucceedBytesUnit(
            FileUtils.readableFileSize(jobDTO.getReadSucceedBytes(), true));
      }
      if (isEditStatus(jobDTO)
          || !processInstanceDto.getId().equals(jobDTO.getProcessInstanceId())) {
        jobDTO.setTaskWriteSpeed(null);
      }

      // 数据迁移，计算表结构迁移进度和数据迁移进度
      if (jobDTO.getJobType() == JobType.TRANSFER.getCode()
          && processInstanceDto.getId().equals(jobDTO.getProcessInstanceId())) {
        jobDTO.setCreateTableProgress(taskTableDetailService.getCreateTableProgressByJobId(jobDTO));
        jobDTO.setDataCollectProgress(taskDataDetailService.getDataCollectProgressByJobId(jobDTO));
      }

      // 数据迁移，仅表结构迁移
      if (jobDTO.getJobType() == JobType.TRANSFER.getCode()
          && jobDTO.getTransferType().equals(TransferType.TABLE.getDescp())) {
        jobDTO.setCreateTableProgress(taskTableDetailService.getCreateTableProgressByJobId(jobDTO));
      }

      // 6定时执行
      if (processInstanceDto.getCommandType() == CommandType.SCHEDULER.getCode()) {
        if (jobDTO.getJobStatus() == TransferStatus.DRAFT.getCode()
            || jobDTO.getJobStatus() == TransferStatus.PRECHECKING.getCode()
            || jobDTO.getJobStatus() == TransferStatus.PRECHECK_FAIL.getCode()) {
          logger.debug("skip");
        } else {
          if (processInstanceDto.getState() == ExecutionStatus.SUCCESS.getCode()
              && jobDTO.getEnableCronTab() == 1) {
            jobDTO.setJobStatus(TransferStatus.TRANSFERING.getCode());
          }
          if (processInstanceDto.getState() == ExecutionStatus.SUCCESS.getCode()
              && jobDTO.getEnableCronTab() == 0) {
            jobDTO.setJobStatus(TransferStatus.COMPLETED.getCode());
          }
          if (processInstanceDto.getState() == ExecutionStatus.RUNNING_EXEUTION.getCode()) {
            jobDTO.setJobStatus(TransferStatus.TRANSFERING.getCode());
          }
          if (processInstanceDto.getState() == ExecutionStatus.FAILURE.getCode()) {
            jobDTO.setJobStatus(TransferStatus.TRANSFER_FAIL.getCode());
          }
          this.updateByPrimaryKeySelective(jobDTO);
        }
      } else {
        if (jobDTO.getJobStatus() == TransferStatus.DRAFT.getCode()
            || jobDTO.getJobStatus() == TransferStatus.PRECHECKING.getCode()
            || jobDTO.getJobStatus() == TransferStatus.PRECHECK_FAIL.getCode()
            || jobDTO.getJobStatus() == TransferStatus.NOT_STARTED.getCode()) {

          logger.debug("skip");
        } else {
          jobDTO.setJobStatus(processInstanceDto.getState());
          this.updateByPrimaryKeySelective(jobDTO);
        }
      }
    } else {
      jobDTO.setTaskAvgSpeedUnit(null);
      jobDTO.setReadSucceedBytesUnit(null);
      jobDTO.setTaskWriteSpeed(null);

      // 数据迁移，计算表结构迁移进度和数据迁移进度
      if (jobDTO.getJobType() == JobType.TRANSFER.getCode()) {
        jobDTO.setCreateTableProgress("0%");
        jobDTO.setDataCollectProgress("0%");
      }
    }

    // 处理速度负数
    if(jobDTO.getTaskWriteSpeed() != null){
      jobDTO.setTaskWriteSpeed(Math.abs(jobDTO.getTaskWriteSpeed()));
    }
    if(jobDTO.getTaskAvgSpeed() != null){
      jobDTO.setTaskAvgSpeed(Math.abs(jobDTO.getTaskAvgSpeed()));
    }
  }

  private boolean isEditStatus(JobDTO jobDTO) {
    return jobDTO.getJobStatus() == TransferStatus.DRAFT.getCode()
        || jobDTO.getJobStatus() == TransferStatus.PRECHECKING.getCode()
        || jobDTO.getJobStatus() == TransferStatus.PRECHECK_FAIL.getCode()
        || jobDTO.getJobStatus() == TransferStatus.NOT_STARTED.getCode();
  }

  /**
   * 验证数据
   *
   * @param jobDTO
   */
  private void checkSaveData(JobDTO jobDTO) {
    CheckParaUtils.checkStringValidity(jobDTO.getName(), logger, "作业名称必填");
    CheckParaUtils.checkObjectValidity(jobDTO.getJobType(), logger, "作业类型必填");

    // 验证源数据源和目标数据源是否相同
    if (JobType.SEMISTRUCTURE.getCode() != jobDTO.getJobType()
        || !"1".equals(jobDTO.getExtraParams())) {
      if (dataSourceUtils.checkSrcAndDstDatasource(
          jobDTO.getSrcDatasourceId(), jobDTO.getDstDatasourceId())) {
        throw new BusinessException(DcsErrorCode.SRC_AND_DST_DATASOURCE_SAME.getDesc());
      }
    }

    // 数据库连接检查
    // 文件采集内置数据源不必检查
    if (JobType.SEMISTRUCTURE.getCode() != jobDTO.getJobType()
            || !"1".equals(jobDTO.getExtraParams())) {
      DataSourceDTO srcDataSource = dataSourceService.queryDataSourceById(jobDTO.getSrcDatasourceId());
      try{
        if (!DBUtils.dataSourceConnTest(srcDataSource)) {
          throwException(srcDataSource.getDatasourceName());
        }
      }catch (Exception e){
        e.printStackTrace();
        logger.error(e.getMessage());
        throwException(srcDataSource.getDatasourceName());
      }
    }

    DataSourceDTO dstDataSource = dataSourceService.queryDataSourceById(jobDTO.getDstDatasourceId());
    try{
      if (!DBUtils.dataSourceConnTest(dstDataSource)) {
        throwException(dstDataSource.getDatasourceName());
      }
    }catch (Exception e){
      logger.error(e.getMessage());
      throwException(dstDataSource.getDatasourceName());
    }
  }

  private void throwException(String datasourceName){
    throw new BusinessException("【" + datasourceName + "】数据源不能连接，请更换数据源");
  }

  /**
   * 新增数据
   *
   * @param jobDTO
   * @throws Exception
   */
  private void insertJob(JobDTO jobDTO) throws Exception {
    jobDTO.setId(ComUtil.getId());
    jobDTO.setEnable(Enable.ONE.getCode());
    jobDTO.setFlag(Enable.ZERO.getCode());
    jobDTO.setJobStatus(TransferStatus.DRAFT.getCode());
    if (JobType.API_HTTP.getCode() != jobDTO.getJobType()) {
      String systemBuilt = jobDTO.getExtraParams();
      jobDTO.setSrcSchemaName(
          dataSourceUtils
              .convertDatasource(jobDTO.getSrcDatasourceId(), systemBuilt)
              .getDatabaseName());
    }
    jobDTO.setDstSchemaName(
        dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId()).getDatabaseName());
    this.insert(jobDTO);

    // 记录日志
    BdpLogUtil.log4Insert(jobDTO);

    // 新增报警信息
    alertService.save(jobDTO.getId(), jobDTO.getJobType());

    // 发送新增消息到kafka
    // sendMsgToKafka("insert" ,jobDTO);
  }

  /**
   * 修改数据
   *
   * @param jobDTO
   * @throws Exception
   */
  private void updateJob(JobDTO jobDTO) throws Exception {
    JobDTO job = this.selectByPrimaryKey(jobDTO.getId());
    CheckParaUtils.checkDbObjectValidity(
        job, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobDTO.getId()));

    // 记录日志
    BdpLogUtil.log4Update(jobDTO, job);

    // 如果源数据源发生改变，需要把selectedTable字段清空
    if (!job.getSrcDatasourceId().equals(jobDTO.getSrcDatasourceId())) {
      jobDTO.setSelectedTable("");
    }

    jobDTO.setTaskAvgSpeed(0L);
    jobDTO.setTaskWriteSpeed(0L);
    jobDTO.setReadSucceedBytes(0L);
    jobDTO.setLastUpdateDate(new Date());
    jobDTO.setJobStatus(TransferStatus.DRAFT.getCode());
    if (JobType.API_HTTP.getCode() != jobDTO.getJobType()) {
      String systemBuilt = jobDTO.getExtraParams();
      jobDTO.setSrcSchemaName(
          dataSourceUtils.convertDatasource(jobDTO.getSrcDatasourceId(), systemBuilt).getDatabaseName());
    }
    jobDTO.setDstSchemaName(
        dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId()).getDatabaseName());
    this.updateByPrimaryKeySelective(jobDTO);
  }

  /** 作业配置第一步(保存数据源) */
  @Transactional(rollbackFor = Exception.class)
  public Map<String, String> saveDataSource(JobDTO jobDTO) throws Exception {
    logger.info("保存数据源,参数为{}", JSONObject.toJSONString(jobDTO));

    // 数据校验
    checkSaveData(jobDTO);

    // 新增或修改
    if (StringUtils.isEmpty(jobDTO.getId())) {
      insertJob(jobDTO);
    } else {
      updateJob(jobDTO);
    }

    Map<String, String> map = new HashMap<>(2);
    map.put("jobId", jobDTO.getId());
    return map;
  }

  /**
   * 作业配置第二步(保存表映射关系)
   *
   * @param newJobDTO 保存新的任务信息
   * @throws Exception 保存异常
   */
  @Transactional(rollbackFor = Exception.class)
  public void saveTableMapping(JobDTO newJobDTO) throws Exception {
    // 判断jobId和对应JOB是否存在
    CheckParaUtils.checkDbObjectValidity(
        newJobDTO.getId(),
        logger,
        String.format(DcsErrorCode.JOB_NULL.getDesc(), newJobDTO.getId()));
    JobDTO oldJobDTO = this.selectByPrimaryKey(newJobDTO.getId());
    CheckParaUtils.checkDbObjectValidity(
        oldJobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), newJobDTO.getId()));

    JobInterface jobInterface = jobFactory.getJobInterface(oldJobDTO.getJobType());
    CheckParaUtils.checkObjectValidity(
        jobInterface, logger, DcsErrorCode.UNSUPPORTED_JOB_TYPE.getDesc() + newJobDTO.getJobType());

    logger.info("保存&更新表映射关系,参数为{}", JSONObject.toJSONString(newJobDTO));

    jobInterface.saveTableMapping(oldJobDTO, newJobDTO);

    tableService.checkTableName(newJobDTO.getId());
  }

  /**
   * 作业配置第四步(更新作业状态)
   *
   * @param jobId
   * @throws Exception
   */
  @Transactional(rollbackFor = Exception.class)
  public void updateJobStatus(String jobId) throws Exception {
    logger.info("更新job状态,jobId:{}", jobId);
    JobDTO jobDTO = this.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    // 1、调用DS保存流程定义数据BusinessExceptionHandler
    int code;
    if (StringUtils.isEmpty(jobDTO.getProcessDefinitionId())) {
      code = taskJsonService.saveProcessDefinition(jobId);
    } else {
      code = taskJsonService.updateProcessDefinition(jobId);
    }
    logger.info("调用DS保存流程定义数据返回结果:{}", code);

    // 2、根据预检查结果更新job状态为：预检失败：22，未启动：23
    int jobStatus;
    if (code == HttpStatus.SC_OK) {
      jobStatus = TransferStatus.NOT_STARTED.getCode();
      List<DcsJobPrecheckDto> preCheckDtoList =
          jobPrecheckResultService.getPrecheckListByJobId(jobId);
      if (CollectionUtils.isNotEmpty(preCheckDtoList)) {
        for (DcsJobPrecheckDto preCheckDto : preCheckDtoList) {
          if (preCheckDto.getResult() == PrecheckResult.FAIL.getCode()) {
            jobStatus = TransferStatus.PRECHECK_FAIL.getCode();
            break;
          }
        }
      }
      jobDTO.setJobStatus(jobStatus);
      jobDTO.setProcessDefinitionId(jobId);
      jobDTO.setTaskAvgSpeed(null);
      jobDTO.setTaskWriteSpeed(null);
      jobDTO.setReadSucceedBytes(null);
      this.updateByPrimaryKey(jobDTO);
      logger.info("修改job状态为:{}", jobStatus);

      // 修改task表is_created字段为0
      TaskDTO taskDTOQuery = new TaskDTO();
      taskDTOQuery.setJobId(jobId);
      taskDTOQuery.setFlag(Enable.ZERO.getCode());
      List<TaskDTO> taskList = taskService.selectList(taskDTOQuery);
      for (TaskDTO task : taskList) {
        task.setIsCreated(Enable.ZERO.getCode());
        taskService.updateByPrimaryKeySelective(task);
      }
    }

    // 发送修改消息到kafka
    // sendMsgToKafka("update" ,jobDTO);
  }

  @Transactional
  public boolean releaseJobByIds(String jobIds, Integer enable) throws Exception {
    String[] idArr = jobIds.split(Constants.COMMA);
    for (String jobId : idArr) {
      JobDTO jobDTO = this.selectByPrimaryKey(jobId);
      CheckParaUtils.checkDbObjectValidity(
          jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));
      jobDTO.setEnable(enable);
      // 启用定时任务，是：1，否：0
      jobDTO.setEnableCronTab(enable);

      // 调ds接口发布任务
      processDefinitionApi.releaseProcessDefinition(jobId, enable);
      this.updateByPrimaryKeySelective(jobDTO);

      // 记录日志
      String logTitle = "启用/停用作业：【" + jobDTO.getName() + "】";
      BdpLogUtil.log("作业管理模块", logTitle, PlatformConstant.OpType.update);
    }
    return true;
  }

  /**
   * 启动作业
   *
   * @param jobIds
   * @return
   * @throws Exception
   */
  @Transactional
  public Boolean startJobByIds(String jobIds) throws Exception {
    String[] idArr = jobIds.split(Constants.COMMA);
    for (String jobId : idArr) {
      JobDTO jobDTO = this.selectByPrimaryKey(jobId);
      CheckParaUtils.checkDbObjectValidity(
          jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));
      CheckParaUtils.checkStringValidity(
          jobDTO.getProcessDefinitionId(), logger, "作业还未创建流程定义，不能启动");

      if (jobDTO.getJobStatus() != TransferStatus.NOT_STARTED.getCode()
          && jobDTO.getJobStatus() != TransferStatus.TRANSFER_FAIL.getCode()
          && jobDTO.getJobStatus() != TransferStatus.COMPLETED.getCode()
          && jobDTO.getJobStatus() != TransferStatus.STOP.getCode()) {
        throw new BusinessException(
            String.format(
                DcsErrorCode.JOB_NOT_START.getDesc(),
                TransferStatus.getEumByCode(jobDTO.getJobStatus()).getDescp()));
      }

      this.precheckHandle(jobDTO);

      // 启动之前先缓存作业与最近一次执行的流程实例
      ProcessInstanceDto processInstanceDto =
          jobDao.getProcessInstanceByProcessDefinitionId(jobDTO.getId());
      if (processInstanceDto != null) {
        jobProcessInstanceMap.put(jobDTO.getId(), processInstanceDto.getId());
      }

      // 1、调ds接口启动任务
      ExecutorQuery executorQuery = new ExecutorQuery();
      executorQuery.setProcessDefinitionId(jobDTO.getId());
      executorQuery.setFailureStrategy(Enable.ZERO.getCode());
      executorQuery.setWarningType(Enable.ZERO.getCode());
      executorQuery.setExecType(Enable.ZERO.getCode());
      executorQuery.setRunMode(Enable.ONE.getCode());

      boolean success = executorApi.startProcessInstance(executorQuery);
      if (success) {
        logger.info("job start success, jobId:{}", jobId);
      } else {
        logger.info("job start failed, jobId:{}", jobId);
        throw new BusinessException(DcsErrorCode.JOB_START_FAIL.getDesc());
      }

      // 2、回写job状态
      jobDTO.setJobStatus(TransferStatus.TRANSFERING.getCode());
      this.updateByPrimaryKeySelective(jobDTO);

      // 记录日志
      String logTitle = "启动作业：【" + jobDTO.getName() + "】";
      BdpLogUtil.log("作业管理模块", logTitle, PlatformConstant.OpType.update);
    }
    return true;
  }

  /**
   * 作业启动前预处理
   *
   * @param jobDTO
   * @throws Exception
   */
  private void precheckHandle(JobDTO jobDTO) throws Exception {
    // 解决手动执行增量同步
    // 先读取流程定义extraParams,如果不为空,则将结束时间修改为当前系统时间
    if (jobDTO.getJobType() == JobType.SYNC.getCode()) {
      TaskDTO taskQuery = new TaskDTO();
      taskQuery.setJobId(jobDTO.getId());
      taskQuery.setFlag(Enable.ZERO.getCode());
      taskQuery.setSyncMode(SyncMode.DATAX_INCREMENT_SYNC.getCode());
      List<TaskDTO> taskList = taskService.selectList(taskQuery);

      ProcessDefinitionDto processDefinitionDto =
          jobDao.getProcessDefinitionByProcessDefinitionId(jobDTO.getId());
      if (processDefinitionDto != null
          && StringUtils.isNotEmpty(processDefinitionDto.getExtraParams())
          && CollectionUtils.isNotEmpty(taskList)
          && taskList.size() > 0) {
        List<Property> properties =
            JSONObject.parseObject(
                processDefinitionDto.getExtraParams(), new TypeReference<List<Property>>() {});
        Map<String, Property> extraParamsMap = this.parseListToMap(properties);
        if (extraParamsMap != null && !extraParamsMap.isEmpty()) {
          // 创建流程定义时，已经放入结束时间参数
          // 将查询数据的结束时间置为下次的查询时间,将系统当前时间作为下次查询的结束时间
          // 目的在于增量同步时，赋值给where条件，区间为左闭右开
          Property endProperty = extraParamsMap.get(DcsConstants.DATA_END_TIME);
          if (endProperty != null && StringUtils.isNotEmpty(endProperty.getValue())) {
            endProperty.setValue(DateUtils.format(new Date(), Constants.YYYY_MM_DD_HH_MM_SS));
            extraParamsMap.put(DcsConstants.DATA_END_TIME, endProperty);
            Collection<Property> values = extraParamsMap.values();
            // 更新流程定义ExtraParams
            jobDao.updateProcessDefinitionExtraParams(
                processDefinitionDto.getId(), JSONObject.toJSONString(values));
            logger.info(
                "update process definition: {}, extraParams: {}",
                processDefinitionDto.getId(),
                extraParamsMap);
          }
        }
      }
    }

    if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
      // 验证目录下文件是否已经全部同步，全部同步，不启动job
      TaskDTO taskDTOQuery = new TaskDTO();
      taskDTOQuery.setJobId(jobDTO.getId());
      taskDTOQuery.setFlag(Enable.ZERO.getCode());
      List<TaskDTO> taskList = taskService.selectList(taskDTOQuery);
      if (CollectionUtils.isNotEmpty(taskList)) {
        int i = 0;
        for (TaskDTO task : taskList) {
          String hdfsFilePath = taskService.getSyncFile(task.getId());
          if (StringUtils.isEmpty(hdfsFilePath)) {
            i++;
          }
        }

        if (taskList.size() == i) {
          throw new BusinessException("该作业下所有文件都已同步");
        }
      }
    }
  }

  /**
   * 删除作业
   *
   * @param jobIds
   * @return
   * @throws Exception
   */
  @Transactional
  public Boolean deleteJobByIds(String jobIds) throws Exception {
    CheckParaUtils.checkStringValidity(jobIds, logger, "没有传入待删除的jobId");

    String[] idArr = jobIds.split(Constants.COMMA);
    for (String jobId : idArr) {
      JobDTO jobDTO = this.selectByPrimaryKey(jobId);
      CheckParaUtils.checkDbObjectValidity(
          jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));
      // 删除流程定义
      try {
        processDefinitionApi.deleteProcessDefinition(jobId);
      } catch (Exception e) {
        logger.error("删除流程定义报错,原因是:" + e.getMessage());
      }

      // 记录日志
      BdpLogUtil.log4Delete(jobDTO);

      jobDTO.setFlag(Enable.ONE.getCode());
      this.updateByPrimaryKeySelective(jobDTO);

      // 发送删除消息到kafka
      // sendMsgToKafka("delete" ,jobDTO);
    }
    return true;
  }

  /*private void sendMsgToKafka(String infoAction, JobDTO jobDTO){
      SendMsgTask task = new SendMsgTask(infoAction, jobDTO);
      threadPool.execute(task);
  }*/

  private void syncSendMsgToKafka(String infoAction, JobDTO jobDTO) {
    try {
      String message = buildKafkaMessage(jobDTO);
      if ("insert".equals(infoAction)) {
        // IntegrationInfoUtils.getInstance().sendInsertMessage(IntegrationInfoUtils.MSG_DCS,
        // message);
      } else if ("update".equals(infoAction)) {
        // IntegrationInfoUtils.getInstance().sendUpdateMessage(IntegrationInfoUtils.MSG_DCS,
        // message);
      } else if ("delete".equals(infoAction)) {
        // IntegrationInfoUtils.getInstance().sendDeleteMessage(IntegrationInfoUtils.MSG_DCS,
        // message);
      }
      logger.info("发送" + infoAction + "消息到kafka成功");
    } catch (Exception e) {
      logger.error("发送" + infoAction + "消息到kafka失败", e);
    }
  }

  private String buildKafkaMessage(JobDTO jobDTO) {
    KafkaJobDto job = new KafkaJobDto();
    job.setId(jobDTO.getId());
    job.setName(jobDTO.getName());
    job.setJob_type(jobDTO.getJobType());
    job.setCreated_by(jobDTO.getCreatedBy());
    job.setCreation_date(jobDTO.getCreationDate());
    job.setLast_updated_by(jobDTO.getLastUpdatedBy());
    job.setLast_update_date(jobDTO.getLastUpdateDate());
    job.setSrc_datasource(buildDatasource(jobDTO.getSrcDatasourceId()));
    job.setDst_datasource(buildDatasource(jobDTO.getDstDatasourceId()));
    job.setSelected_table(buildTableMapping(jobDTO.getId()));
    return JSONUtils.toJson(job);
  }

  private List<TableMappingDto> buildTableMapping(String jobId) {
    List<TableMappingDto> mappingList = new ArrayList<>();
    List<TaskDTO> taskList = taskService.getTaskListByJobId(jobId);
    if (CollectionUtils.isNotEmpty(taskList)) {
      for (TaskDTO task : taskList) {
        TableMappingDto dto = new TableMappingDto();
        dto.setSrc_table(task.getSrcTableName());
        dto.setDst_table(task.getDstTableName());
        List<ColumnMappingDto> columnInfoDtoList =
            JSONUtils.toList(task.getMappingColumn(), ColumnMappingDto.class);
        dto.setMapping_column(columnInfoDtoList);
        mappingList.add(dto);
      }
    }
    return mappingList;
  }

  private KafkaJobDto.KafkaDatasource buildDatasource(String datasourceId) {
    KafkaJobDto.KafkaDatasource datasource = new KafkaJobDto().new KafkaDatasource();
    DataSourceDTO dataSourceDTO = dataSourceService.selectByPrimaryKey(datasourceId);
    if (dataSourceDTO != null) {
      datasource.setId(dataSourceDTO.getId());
      datasource.setName(dataSourceDTO.getDatasourceName());
      datasource.setType(dataSourceDTO.getDatasourceType());
      datasource.setType_id(dataSourceDTO.getTypeId());
      datasource.setConnection_params(dataSourceDTO.getConnectionParams());
      datasource.setStructure(dataSourceDTO.getDatasourceStructure());
      datasource.setProject_type(dataSourceDTO.getProjectType());
    }
    return datasource;
  }

  @Transactional
  public void deleteJob() throws Exception {
    JobDTO jobQuery = new JobDTO();
    jobQuery.setFlag(Enable.ONE.getCode());
    List<JobDTO> jobList = this.selectList(jobQuery);

    if (CollectionUtils.isEmpty(jobList)) {
      return;
    }

    for (JobDTO jobDTO : jobList) {
      // 物理删除
      // 1、删除bdp_dcs_job
      this.deleteByPrimaryKey(jobDTO.getId());

      // 2、删除bdp_dcs_alert
      AlertDTO alertDTOQuery = new AlertDTO();
      alertDTOQuery.setJobId(jobDTO.getId());
      List<AlertDTO> alertList = alertService.selectList(alertDTOQuery);
      if (CollectionUtils.isNotEmpty(alertList)) {
        for (AlertDTO alert : alertList) {
          alertService.deleteByPrimaryKey(alert.getId());
        }
      }

      // 3、删除bdp_dcs_job_precheck_result
      JobPrecheckResultDTO jobPrecheckResultDTOQuery = new JobPrecheckResultDTO();
      jobPrecheckResultDTOQuery.setJobId(jobDTO.getId());
      List<JobPrecheckResultDTO> jobPrecheckResultList =
          jobPrecheckResultService.selectList(jobPrecheckResultDTOQuery);
      if (CollectionUtils.isNotEmpty(jobPrecheckResultList)) {
        for (JobPrecheckResultDTO jobPrecheckResult : jobPrecheckResultList) {
          jobPrecheckResultService.deleteByPrimaryKey(jobPrecheckResult.getId());
        }
      }

      // 5、删除 bdp_dcs_table_rule
      TableRuleDTO tableRuleDTOQuery = new TableRuleDTO();
      tableRuleDTOQuery.setJobId(jobDTO.getId());
      List<TableRuleDTO> tableRuleList = tableRuleService.selectList(tableRuleDTOQuery);
      if (CollectionUtils.isNotEmpty(tableRuleList)) {
        for (TableRuleDTO tableRule : tableRuleList) {
          tableRuleService.deleteByPrimaryKey(tableRule.getId());
        }
      }

      // 删除bdp_dcs_job_file
      JobFileDTO jobFileDTOQuery = new JobFileDTO();
      jobFileDTOQuery.setJobId(jobDTO.getId());
      List<JobFileDTO> jobFileList = jobFileService.selectList(jobFileDTOQuery);
      if (CollectionUtils.isNotEmpty(jobFileList)) {
        for (JobFileDTO jobFile : jobFileList) {
          jobFileService.deleteByPrimaryKey(jobFile.getId());
        }
      }

      // 6、删除bdp_dcs_task
      TaskDTO taskDTOQuery = new TaskDTO();
      taskDTOQuery.setJobId(jobDTO.getId());
      List<TaskDTO> taskList = taskService.selectList(taskDTOQuery);
      if (CollectionUtils.isNotEmpty(taskList)) {
        for (TaskDTO task : taskList) {
          taskService.deleteByPrimaryKey(task.getId());

          // 7、删除bdp_dcs_task_table_detail
          TaskTableDetailDTO taskTableDetailDTOQuery =
              TaskTableDetailDTO.builder().taskId(task.getId()).build();
          List<TaskTableDetailDTO> taskTableDetailList =
              taskTableDetailService.selectList(taskTableDetailDTOQuery);
          if (CollectionUtils.isNotEmpty(taskTableDetailList)) {
            for (TaskTableDetailDTO taskTableDetail : taskTableDetailList) {
              taskTableDetailService.deleteByPrimaryKey(taskTableDetail.getId());
            }
          }

          // 8、删除bdp_dcs_task_data_detail
          TaskDataDetailDTO taskDataDetailDTOQuery =
              TaskDataDetailDTO.builder().taskId(task.getId()).build();
          List<TaskDataDetailDTO> taskDataDetailList =
              taskDataDetailService.selectList(taskDataDetailDTOQuery);
          if (CollectionUtils.isNotEmpty(taskDataDetailList)) {
            for (TaskDataDetailDTO taskDataDetail : taskDataDetailList) {
              taskDataDetailService.deleteByPrimaryKey(taskDataDetail.getId());
            }
          }

          // 11、删除bdp_dcs_column_rule
          ColumnRuleDTO columnRuleDTOQuery = new ColumnRuleDTO();
          columnRuleDTOQuery.setTaskId(task.getId());
          List<ColumnRuleDTO> columnRuleList = columnRuleService.selectList(columnRuleDTOQuery);
          if (CollectionUtils.isNotEmpty(columnRuleList)) {
            for (ColumnRuleDTO columnRule : columnRuleList) {
              columnRuleService.deleteByPrimaryKey(columnRule.getId());
            }
          }
        }
      }

      // 12、删除流程定义
      try {
        processDefinitionApi.deleteProcessDefinition(jobDTO.getId());
      } catch (Exception e) {
        logger.error("删除流程定义报错,原因是:" + e.getMessage());
      }
    }
  }

  /**
   * 拷贝作业
   *
   * @param jobId
   * @param name
   * @return
   */
  @Transactional
  public Boolean copyJobById(String jobId, String name) {
    JobDTO jobDTO = this.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));
    CheckParaUtils.checkStringValidity(name, logger, "job名称必填");

    // 1、拷贝job表
    String newJobId = ComUtil.getId();
    jobDTO.setId(newJobId);
    jobDTO.setName(name);
    jobDTO.setJobStatus(TransferStatus.DRAFT.getCode());
    jobDTO.setTaskAvgSpeed(null);
    jobDTO.setTaskWriteSpeed(null);
    jobDTO.setReadSucceedBytes(null);
    jobDTO.setProcessDefinitionId(null);
    this.insert(jobDTO);

    // 2、拷贝bdp_dcs_alert表
    AlertDTO alertDTOQuery = new AlertDTO();
    alertDTOQuery.setJobId(jobId);
    List<AlertDTO> alertList = alertService.selectList(alertDTOQuery);
    if (CollectionUtils.isNotEmpty(alertList)) {
      for (AlertDTO alert : alertList) {
        alert.setId(ComUtil.getId());
        alert.setJobId(newJobId);
        alert.setAlertStatus(AlertStatus.ENABLE.getCode());
        alertService.insert(alert);
      }
    }

    // 3、拷贝bdp_dcs_job_precheck_result表
    JobPrecheckResultDTO jobPrecheckResultDTOQuery = new JobPrecheckResultDTO();
    jobPrecheckResultDTOQuery.setJobId(jobId);
    List<JobPrecheckResultDTO> jobPrecheckResultList =
        jobPrecheckResultService.selectList(jobPrecheckResultDTOQuery);
    if (CollectionUtils.isNotEmpty(jobPrecheckResultList)) {
      for (JobPrecheckResultDTO jobPrecheckResult : jobPrecheckResultList) {
        jobPrecheckResult.setId(ComUtil.getId());
        jobPrecheckResult.setJobId(newJobId);
        jobPrecheckResult.setResult(PrecheckResult.WAIT.getCode());
        jobPrecheckResultService.insert(jobPrecheckResult);
      }
    }

    // 4、拷贝bdp_dcs_table_rule表
    TableRuleDTO tableRuleDTOQuery = new TableRuleDTO();
    tableRuleDTOQuery.setJobId(jobId);
    List<TableRuleDTO> tableRuleList = tableRuleService.selectList(tableRuleDTOQuery);
    if (CollectionUtils.isNotEmpty(tableRuleList)) {
      for (TableRuleDTO tableRule : tableRuleList) {
        tableRule.setId(ComUtil.getId());
        tableRule.setJobId(newJobId);
        tableRuleService.insert(tableRule);
      }
    }

    // 5、拷贝bdp_dcs_task表
    TaskDTO taskDTOQuery = new TaskDTO();
    taskDTOQuery.setJobId(jobId);
    taskDTOQuery.setFlag(Enable.ZERO.getCode());
    List<TaskDTO> taskList = taskService.selectList(taskDTOQuery);
    if (CollectionUtils.isNotEmpty(taskList)) {
      for (TaskDTO task : taskList) {
        String oldTaskId = task.getId();
        String newTaskId = ComUtil.getId();
        task.setId(newTaskId);
        task.setJobId(newJobId);
        taskService.insert(task);

        // 6、拷贝bdp_dcs_column_rule表
        ColumnRuleDTO columnRuleDTOQuery = new ColumnRuleDTO();
        columnRuleDTOQuery.setTaskId(oldTaskId);
        List<ColumnRuleDTO> columnRuleList = columnRuleService.selectList(columnRuleDTOQuery);
        if (CollectionUtils.isNotEmpty(columnRuleList)) {
          for (ColumnRuleDTO columnRule : columnRuleList) {
            columnRule.setId(ComUtil.getId());
            columnRule.setTaskId(newTaskId);
            columnRuleService.insert(columnRule);
          }
        }

        // 7、拷贝bdp_dcs_task_table_detail表
        TaskTableDetailDTO taskTableDetailDTOQuery =
            TaskTableDetailDTO.builder().taskId(oldTaskId).build();
        List<TaskTableDetailDTO> taskTableDetailList =
            taskTableDetailService.selectList(taskTableDetailDTOQuery);
        if (CollectionUtils.isNotEmpty(taskTableDetailList)) {
          for (TaskTableDetailDTO taskTableDetail : taskTableDetailList) {
            taskTableDetail.setId(ComUtil.getId());
            taskTableDetail.setTaskId(newTaskId);
            taskTableDetail.setTaskStatus(TaskDetailStatus.NOT_STARTED.getCode());
            taskTableDetailService.insert(taskTableDetail);
          }
        }

        // 8、拷贝bdp_dcs_task_data_detail表
        TaskDataDetailDTO taskDataDetailDTOQuery =
            TaskDataDetailDTO.builder().taskId(oldTaskId).build();
        List<TaskDataDetailDTO> taskDataDetailList =
            taskDataDetailService.selectList(taskDataDetailDTOQuery);
        if (CollectionUtils.isNotEmpty(taskDataDetailList)) {
          for (TaskDataDetailDTO taskDataDetail : taskDataDetailList) {
            taskDataDetail.setId(ComUtil.getId());
            taskDataDetail.setTaskId(newTaskId);
            taskDataDetail.setTaskStatus(TaskDetailStatus.NOT_STARTED.getCode());
            taskDataDetail.setEstimateTotal(null);
            taskDataDetail.setCompletedTotal(null);
            taskDataDetail.setCostTime(null);
            taskDataDetailService.insert(taskDataDetail);
          }
        }
      }
    }
    return true;
  }

  /**
   * 查看作业详情(第一步)
   *
   * @param jobId
   * @return
   */
  public JobDTO getDataSourceByJobId(String jobId) {
    JobDTO job = this.selectByPrimaryKey(jobId);
    BdpLogUtil.log4Query(job);
    return job;
  }

  /** 查看作业详情(第二步) */
  public Map<String, Object> getTableMappingByJobId(String jobId) throws Exception {
    Map<String, Object> map = new HashMap<>();
    JobDTO jobDTO = this.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    if (jobDTO.getJobType() == JobType.TRANSFER.getCode()
        || jobDTO.getJobType() == JobType.SYNC.getCode()) {
      String selectedTable = jobDTO.getSelectedTable();
      if (StringUtils.isEmpty(selectedTable)) {
        return null;
      }

      List<Map<String, Object>> dstTableList = new ArrayList<>();
      // 根据数据源id获取所有表名
      List<String> tableList = baseDbService.getTableList(jobDTO.getDstDatasourceId());
      List<TableMappingDto> tableMappingDtoList =
          JSONUtils.toList(selectedTable, TableMappingDto.class);
      if (CollectionUtils.isNotEmpty(tableMappingDtoList)) {
        for (TableMappingDto tableMappingDto : tableMappingDtoList) {
          boolean sameTableName = false;
          Map<String, Object> tableMappingMap = new HashMap<>();
          if (tableList.contains(tableMappingDto.getDst_table())) {
            sameTableName = true;
          }
          tableMappingMap.put("id", "");
          tableMappingMap.put("dstTable", tableMappingDto.getDst_table());
          tableMappingMap.put("sameName", sameTableName ? 1 : 0);
          dstTableList.add(tableMappingMap);
        }
      }
      List<TableRuleDTO> tableRuleList = tableRuleService.getTableRuleListByJobId(jobId);
      map.put("dstTableList", dstTableList);
      map.put("tableRuleList", tableRuleList);
    }

    map.put("jobDTO", jobDTO);
    return map;
  }

  /**
   * 查看作业详情(第三步获取目标表)
   *
   * @param jobId
   * @return
   * @throws Exception
   */
  public List<Map<String, Object>> getDstTableByJobId(String jobId) throws Exception {
    JobDTO jobDTO = this.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    String selectedTable = jobDTO.getSelectedTable();
    if (StringUtils.isEmpty(selectedTable)) {
      return null;
    }

    List<Map<String, Object>> dstTableList = new ArrayList<>();
    // 根据数据源id获取所有表名
    List<String> tableList = baseDbService.getTableList(jobDTO.getDstDatasourceId());
    List<TableMappingDto> tableMappingDtoList =
        JSONUtils.toList(selectedTable, TableMappingDto.class);
    if (CollectionUtils.isNotEmpty(tableMappingDtoList)) {
      for (TableMappingDto tableMappingDto : tableMappingDtoList) {
        boolean sameTableName = false;
        Map<String, Object> tableMappingMap = new HashMap<>();
        if (tableList.contains(tableMappingDto.getDst_table())) {
          sameTableName = true;
        }
        tableMappingMap.put("id", "");
        tableMappingMap.put("srcTable", tableMappingDto.getSrc_table());
        tableMappingMap.put("dstTable", tableMappingDto.getDst_table());
        tableMappingMap.put("sameName", sameTableName ? 1 : 0);
        dstTableList.add(tableMappingMap);
      }
    }
    return dstTableList;
  }

  /**
   * 检查第三步配置数据
   *
   * @param jobDTO
   * @return
   * @throws Exception
   */
  @Transactional
  public Boolean checkTaskConfig(JobDTO jobDTO) throws Exception {
    String jobId = jobDTO.getId();
    CheckParaUtils.checkStringValidity(jobId, logger, "jobId不能为空");
    JobDTO job = this.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(
        job, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    StringBuilder builder = new StringBuilder();
    TaskDTO taskDTOQuery = new TaskDTO();
    taskDTOQuery.setJobId(jobId);
    taskDTOQuery.setFlag(Enable.ZERO.getCode());
    List<TaskDTO> taskList = taskService.selectList(taskDTOQuery);
    if (CollectionUtils.isNotEmpty(taskList)) {
      for (TaskDTO task : taskList) {
        // 生成DataxJson(为了防止第三步用户直接点击下一步，此时dataxJson可能还未生成)
        String dataxJson = dataxJsonService.buildDataxJson(task);
        task.setDataxJson(dataxJson);
        taskService.updateByPrimaryKeySelective(task);

        if (task.getMappingColumn() == null
            || task.getDataxJson() == null
            || "[]".equals(task.getMappingColumn())) {
          String srcTableName = null;
          if (task.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
            FileTemplateDTO fileTemplate =
                fileTemplateService.selectByPrimaryKey(task.getSrcTableName());
            CheckParaUtils.checkDbObjectValidity(
                fileTemplate,
                logger,
                String.format("没有查到数据，请确保templateId：{}的模板是否存在", task.getSrcTableName()));
            srcTableName = fileTemplate.getName();
          } else {
            srcTableName = task.getSrcTableName();
          }
          builder.append(
              String.format(DcsErrorCode.TASK_NOT_CONFIG_COLUMN_MAPPING.getDesc(), srcTableName));
        }

        // 检查增量字段是否设置
        structureTaskService.checkSync(task.getJobType(), task.getSyncMode(),
                task.getFieldName(), task.getSyncTimePoint(), task.getSrcTableName());

        try {
          String mappingColumn = task.getMappingColumn();
          JSONObject.parseObject(dataxJson);

          if (task.getJobType() == JobType.TRANSFER.getCode()
              || task.getJobType() == JobType.SYNC.getCode()
              || task.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
            JSONUtils.toList(mappingColumn, ColumnMappingDto.class);
          } else if (task.getJobType() == JobType.STREAM.getCode()) {
            List<StreamColumnMappingDto> streamColumnMappingDtoList =
                JSONUtils.toList(mappingColumn, StreamColumnMappingDto.class);

            CheckParaUtils.checkCollectionValidity(
                streamColumnMappingDtoList,
                logger,
                String.format(
                    DcsErrorCode.TASK_NOT_CONFIG_COLUMN_MAPPING.getDesc(), task.getSrcTableName()));

            for (StreamColumnMappingDto streamColumnMappingDto : streamColumnMappingDtoList) {
              CheckParaUtils.checkCollectionValidity(
                  streamColumnMappingDto.getColumnList(),
                  logger,
                  String.format(
                      DcsErrorCode.TASK_NOT_CONFIG_COLUMN_MAPPING.getDesc(),
                      task.getSrcTableName()));
            }
          }
        } catch (Exception e) {
          logger.error("配置检查失败,原因是:{}", e.getMessage());
          logger.error("config is unvalid json, please check taskId is {}", task.getId());
          return false;
        }
      }
    }

    if (StringUtils.isNotEmpty(builder.toString())) {
      throw new BusinessException(builder.toString());
    }
    return true;
  }

  /**
   * 停止、暂停作业
   *
   * @param jobIds
   * @param executeType
   * @return
   */
  @Transactional
  public Boolean excuteJobByIds(String jobIds, String executeType) {
    String[] idArr = jobIds.split(Constants.COMMA);
    for (String jobId : idArr) {
      JobDTO jobDTO = this.selectByPrimaryKey(jobId);
      CheckParaUtils.checkDbObjectValidity(
          jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

      // 1、调ds接口执行暂停，终止采集任务
      boolean success = executorApi.execute(jobDTO.getId(), executeType);
      if (success) {
        logger.info("job excute success, jobId:{}", jobId);
      } else {
        logger.info("job execute failed, jobId:{}", jobId);
        throw new BusinessException(
            String.format(
                "job execute failed,executeType is " + "{}, jobId is {}", executeType, jobId));
      }
      int status =
          DcsConstants.STOP.equals(executeType)
              ? TransferStatus.NOT_STARTED.getCode()
              : TransferStatus.SUSPEND.getCode();
      jobDTO.setJobStatus(status);
      this.updateByPrimaryKeySelective(jobDTO);

      // 2、更新bdp_dcs_task_table_detail表状态字段
      List<TableDetailDto> taskTableDetailList =
          taskTableDetailService.getTaskTableDetailListByJobId(jobId);
      if (CollectionUtils.isNotEmpty(taskTableDetailList)) {
        for (TableDetailDto tableDetail : taskTableDetailList) {
          TaskTableDetailDTO taskTableDetailDTO =
              taskTableDetailService.selectByPrimaryKey(tableDetail.getId());
          taskTableDetailDTO.setTaskStatus(status);
          taskTableDetailService.updateByPrimaryKeySelective(taskTableDetailDTO);
        }
      }

      // 记录日志
      String logTitle = "暂停/结束作业：【" + jobDTO.getName() + "】";
      BdpLogUtil.log("作业管理模块", logTitle, PlatformConstant.OpType.update);
    }
    return true;
  }

  /**
   * 查询作业下执行的任务列表
   *
   * @param processInstanceId
   * @param currentPage
   * @param pageSize
   * @return
   */
  public QueryRespBean<TaskSpeedDTO> getHistoryTaskList(
      String processInstanceId, Integer currentPage, Integer pageSize) {
    ProcessInstanceDto processInstance = getProcessInstanceById(processInstanceId);
    CheckParaUtils.checkObjectValidity(
        processInstance, logger, "流程实例为空,processInstanceId:" + processInstanceId);

    List<String> processInstanceList = new ArrayList<>();
    processInstanceList.add(processInstanceId);
    Map<String, String> taskTypeMap = getProcessTypeMap(processInstanceList);

    if (DcsConstants.SPARK.equals(taskTypeMap.get(processInstanceId))) {
      return getSparkHistoryTaskList(processInstanceId, currentPage, pageSize);
    } else {
      return getDataXHistoryTaskList(processInstanceId, currentPage, pageSize);
    }
  }

  /**
   * 查询SPARK任务信息
   *
   * @param processInstanceId
   * @param currentPage
   * @param pageSize
   * @return
   */
  private QueryRespBean<TaskSpeedDTO> getSparkHistoryTaskList(
      String processInstanceId, Integer currentPage, Integer pageSize) {
    QueryRespBean<TaskSpeedDTO> queryRespBean = new QueryRespBean<>();
    com.github.pagehelper.PageHelper.startPage(currentPage, pageSize);
    Page<TaskSpeedDTO> historyTaskSpeedDtoListQuery =
        taskLogService.getHistorySparkTaskListByProcessInstanceId(processInstanceId);

    List<TaskSpeedDTO> result = historyTaskSpeedDtoListQuery.getResult();
    result.forEach(
        taskSpeedDTO -> {
          try {
            taskSpeedDTO.setCostTime(taskSpeedDTO.getStartTimeStamp() + "s");
            taskSpeedDTO.setName(
                taskSpeedDTO.getSrcTableName() + "->" + taskSpeedDTO.getDstTableName());
          } catch (Exception e) {
            e.printStackTrace();
            logger.error(e.getMessage());
          }
        });
    queryRespBean.setResult(historyTaskSpeedDtoListQuery);
    return queryRespBean;
  }

  /**
   * 查询DATAX任务信息
   *
   * @param processInstanceId
   * @param currentPage
   * @param pageSize
   * @return
   */
  private QueryRespBean<TaskSpeedDTO> getDataXHistoryTaskList(
      String processInstanceId, Integer currentPage, Integer pageSize) {
    QueryRespBean<TaskSpeedDTO> queryRespBean = new QueryRespBean<>();
    com.github.pagehelper.PageHelper.startPage(currentPage, pageSize);
    Page<TaskSpeedDTO> historyTaskSpeedDtoListQuery =
        taskLogService.getHistoryTaskSpeedListByProcessInstanceId(processInstanceId);

    List<TaskSpeedDTO> result = historyTaskSpeedDtoListQuery.getResult();
    if (CollectionUtils.isNotEmpty(result)) {
      for (TaskSpeedDTO taskSpeedDTO : result) {
        try {
          this.handleResult(taskSpeedDTO);
        } catch (Exception e) {
          e.printStackTrace();
          logger.error(e.getMessage());
        }
      }
    }
    queryRespBean.setResult(historyTaskSpeedDtoListQuery);
    return queryRespBean;
  }

  /**
   * 计算、格式化结果数据
   *
   * @param taskSpeedDTO
   */
  private void handleResult(TaskSpeedDTO taskSpeedDTO) {
    // 半结构化采集源表为模板ID
    if (taskSpeedDTO.getJobType() != null
        && taskSpeedDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
      JobFileDTO jobFileDTOQuery = new JobFileDTO();
      jobFileDTOQuery.setTaskId(taskSpeedDTO.getTaskId());
      jobFileDTOQuery.setProcessInstanceId(taskSpeedDTO.getProcessInstanceId());
      JobFileDTO jobFileDTO = jobFileService.selectOne(jobFileDTOQuery);
      if (jobFileDTO != null) {
        taskSpeedDTO.setSrcTableName(jobFileDTO.getFilePath());
      } else {
        FileTemplateDTO templateDTO =
            fileTemplateService.selectByPrimaryKey(taskSpeedDTO.getSrcTableName());
        CheckParaUtils.checkDbObjectValidity(
            templateDTO,
            logger,
            String.format("没有查到数据，请确保templateId：{}的模板是否存在", taskSpeedDTO.getSrcTableName()));
        taskSpeedDTO.setSrcTableName(templateDTO.getName());
      }
    }

    taskSpeedDTO.setName(taskSpeedDTO.getSrcTableName() + "->" + taskSpeedDTO.getDstTableName());
    Long startTimestamp = taskSpeedDTO.getStartTimeStamp();
    Long endTimestamp = taskSpeedDTO.getEndTimeStamp();
    if (startTimestamp == null || endTimestamp == null) {
      throw new BusinessException(
          String.format("任务开始或者结束时间不能为空，请检查！ 任务实例id为:{}", taskSpeedDTO.getId()));
    }

    taskSpeedDTO.setStartTime(DateUtils.getDateByString(new Date(startTimestamp)));
    taskSpeedDTO.setEndTime(DateUtils.getDateByString(new Date(endTimestamp)));

    long totalTime = endTimestamp - startTimestamp;
    long sec = totalTime < 1000 ? 1 : totalTime / 1000;
    taskSpeedDTO.setCostTime(sec + "s");
    long totalReadRecords = taskSpeedDTO.getReadSucceedRecords();
    // B/s
    long avgReadBytes = Math.round((double) taskSpeedDTO.getReadSucceedBytes() / sec);
    long avgWriteRecords = taskSpeedDTO.getReadSucceedRecords() / sec;
    taskSpeedDTO.setTaskAvgSpeed(FileUtils.readableSpeedSize(avgReadBytes));
    taskSpeedDTO.setTaskWriteSpeed(avgWriteRecords + "rec/s");
    taskSpeedDTO.setReadSuccessBytesString(
        FileUtils.readableFileSize(taskSpeedDTO.getReadSucceedBytes(), true));
    taskSpeedDTO.setReadSucceedRecords(totalReadRecords);

    DecimalFormat df = new DecimalFormat(DcsConstants.PATTERN);
    if (taskSpeedDTO.getReadSucceedRecords() != null) {
      taskSpeedDTO.setReadSucceedRecordsString(df.format(taskSpeedDTO.getReadSucceedRecords()));
    }

    // 正在运行中，结束时间为空
    if (taskSpeedDTO.getExecuteStatus() == SyncStatus.SYNCRING.getCode()) {
      taskSpeedDTO.setEndTime("---");
    }
  }

  /**
   * 保存&修改定时任务
   *
   * @param processSchedulerDto
   * @return
   */
  @Transactional(rollbackFor = Exception.class)
  public Boolean saveSchedule(ProcessSchedulerDto processSchedulerDto) {
    logger.info("保存定时任务接口入参：{}", processSchedulerDto.toString());
    CheckParaUtils.checkStringValidity(
        processSchedulerDto.getProcessDefinitionId(), logger, "流程定义id不能为空！");
    CheckParaUtils.checkObjectValidity(processSchedulerDto.getStartTime(), logger, "开始时间不能为空！");
    CheckParaUtils.checkObjectValidity(processSchedulerDto.getEndTime(), logger, "结束时间不能为空！");
    CheckParaUtils.checkStringValidity(
        processSchedulerDto.getCrontab(), logger, "crontab时间表达式不能为空！");
    JobDTO jobDTO = this.selectByPrimaryKey(processSchedulerDto.getProcessDefinitionId());
    CheckParaUtils.checkDbObjectValidity(
        jobDTO,
        logger,
        String.format(
            DcsErrorCode.JOB_NULL.getDesc(), processSchedulerDto.getProcessDefinitionId()));

    // 判断这个流程下的定时任务个数，如果有则修改，没有则添加
    List<Map<String, Object>> result =
        this.getScheduleListByJobId(processSchedulerDto.getProcessDefinitionId());
    boolean success;
    if (CollectionUtils.isEmpty(result)) {
      success = schedulerApi.createSchedule(processSchedulerDto);
      if (success) {
        jobDTO.setConfigCronTab(Enable.ONE.getCode());
        this.updateByPrimaryKeySelective(jobDTO);
        logger.info("新增定时任务成功, {}", processSchedulerDto);
      } else {
        logger.info("新增定时任务失败, {}", processSchedulerDto);
      }
    } else {
      CheckParaUtils.checkStringValidity(processSchedulerDto.getId(), logger, "定时设置ID不能为空");
      success = schedulerApi.updateSchedule(processSchedulerDto);
      if (success) {
        logger.info("修改定时任务成功, {}", processSchedulerDto);
      } else {
        logger.info("修改定时任务失败, {}", processSchedulerDto);
      }
    }
    return success;
  }

  /**
   * 根据流程id获取定时任务列表
   *
   * @param jobId 流程id
   * @return 定时任务列表
   */
  public List<Map<String, Object>> getScheduleListByJobId(String jobId) {
    QueryRespBean<Map<String, Object>> mapQueryRespBean =
            schedulerApi.queryScheduleListPaging(jobId, null, 1, 10);
    return mapQueryRespBean.getResult().getResult();
  }

  /**
   * 根据jobId查看定时任务用来页面回显
   *
   * @param jobId
   * @return
   */
  public Map<String, Object> getScheduleByJobId(String jobId) {
    List<Map<String, Object>> list = this.getScheduleListByJobId(jobId);
    if (CollectionUtils.isEmpty(list)) {
      return null;
    }

    Map<String, Object> map = list.get(0);
    map.put("failureStrategy", map.get("failureStrategy"));
    map.put("warningType", map.get("warningType"));
    map.put("processInstancePriority", map.get("processInstancePriority"));
    return map;
  }

  /**
   * 查询历史作业
   *
   * @param jobId
   * @param currentPage
   * @param pageSize
   * @return
   */
  public QueryRespBean<JobLogDTO> getHistoryJobList(
      String jobId, Integer currentPage, Integer pageSize) {
    com.github.pagehelper.PageHelper.startPage(currentPage, pageSize);
    Page<JobLogDTO> historyJobLogList = taskLogService.getHistoryJobLogListByJobId(jobId);
    List<JobLogDTO> jobList = historyJobLogList.getResult();

    List<String> processInstanceList =
        jobList.stream().map(JobLogDTO::getProcessInstanceId).collect(Collectors.toList());
    Map<String, Long> sourceDiffMap =
        taskLogService.getSourceDiffListByProcessInstance(processInstanceList);

    Map<String, String> taskTypeMap = getProcessTypeMap(processInstanceList);

    if (CollectionUtils.isNotEmpty(jobList)) {
      DecimalFormat df = new DecimalFormat(DcsConstants.PATTERN);

      for (JobLogDTO jobLogDTO : jobList) {
        jobLogDTO.setTaskType(taskTypeMap.get(jobLogDTO.getProcessInstanceId()));

        handleSparkTask(jobLogDTO, sourceDiffMap);

        if (jobLogDTO.getReadSucceedRecords() != null) {
          jobLogDTO.setReadSucceedRecordsString(df.format(jobLogDTO.getReadSucceedRecords()));
        }
        if (jobLogDTO.getReadSucceedBytes() != null) {
          jobLogDTO.setReadSuccessBytesString(
              FileUtils.readableFileSize(jobLogDTO.getReadSucceedBytes(), true));
        }
        if (jobLogDTO.getJobAvgSpeed() != null) {
          jobLogDTO.setJobAvgSpeedString(FileUtils.readableSpeedSize(jobLogDTO.getJobAvgSpeed()));
        }
        if (jobLogDTO.getStartTime() != null) {
          jobLogDTO.setStartTimeString(DateUtils.getDateByString(jobLogDTO.getStartTime()));
        }
        if (jobLogDTO.getEndTime() != null) {
          jobLogDTO.setEndTimeString(DateUtils.getDateByString(jobLogDTO.getEndTime()));
        }
      }
    }

    QueryRespBean<JobLogDTO> queryRespBean = new QueryRespBean<>();
    queryRespBean.setResult(historyJobLogList);
    BdpLogUtil.log4Query(queryRespBean);
    return queryRespBean;
  }

  private void handleSparkTask(JobLogDTO jobLogDTO, Map<String, Long> sourceDiffMap) {
    if (DcsConstants.SPARK.equals(jobLogDTO.getTaskType())) {
      jobLogDTO.setReadSucceedRecords(sourceDiffMap.get(jobLogDTO.getProcessInstanceId()));

      if (jobLogDTO.getReadSucceedRecords() != null) {
        Long jobWriteSpeed = jobLogDTO.getReadSucceedRecords() / jobLogDTO.getCostTime();
        jobLogDTO.setJobWriteSpeed(jobWriteSpeed);
      }
    }
  }

  /**
   * 获取任务类型SPARK DATAX
   *
   * @param
   * @return
   */
  private Map<String, String> getProcessTypeMap(List<String> processInstanceList) {
    Map<String, String> map = new HashMap<>(16);
    if (CollectionUtils.isNotEmpty(processInstanceList)) {
      List<TaskInstanceDto> taskInstanceList =
          jobDao.getTaskInstanceListByProcessInstanceId(processInstanceList);

      if (CollectionUtils.isNotEmpty(taskInstanceList)) {
        for (TaskInstanceDto taskInstance : taskInstanceList) {
          map.put(taskInstance.getProcessInstanceId(), taskInstance.getTaskType());
        }
      }
    }
    return map;
  }

  /**
   * 预览任务调度执行时间点
   *
   * @param scheduleParamDto
   * @return
   */
  public List<String> previewSchedule(ScheduleParamDto scheduleParamDto) {
    List<String> list = new ArrayList<>();
    List<Date> dateList = schedulerApi.previewSchedule(scheduleParamDto);
    if (CollectionUtils.isNotEmpty(dateList)) {
      for (Date date : dateList) {
        list.add(DateUtils.getDateByString(date));
      }
    } else {
      throw new BusinessException(
          String.format(
              "未来执行时间不在时间范围内[{}-{}]",
              DateUtils.format(scheduleParamDto.getStartTime(), Constants.YYYY_MM_DD_HH_MM_SS),
              DateUtils.format(scheduleParamDto.getEndTime(), Constants.YYYY_MM_DD_HH_MM_SS)));
    }
    return list;
  }

  private Map<String, Property> parseListToMap(List<Property> list) {
    if (CollectionUtils.isNotEmpty(list)) {
      Map<String, Property> map = new LinkedHashMap<>();
      for (Property property : list) {
        map.put(property.getProp(), property);
      }
      return map;
    }
    return null;
  }

  public List<SyncFileDto> getDstFileByJobId(String jobId) throws Exception {
    JobDTO jobDTO = this.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    String selectedFile = jobDTO.getSelectedFile();
    if (StringUtils.isEmpty(selectedFile)) {
      return null;
    }

    return JSONUtils.toList(selectedFile, SyncFileDto.class);
  }

  public TaskInstanceDto getTaskInstanceById(String taskInstanceId) {
    return jobDao.getTaskInstanceById(taskInstanceId);
  }

  public TaskInstanceDto getTaskInstanceByProcessInstanceId(String processInstanceId, String name) {
    return jobDao.getTaskInstanceByProcessInstanceId(processInstanceId, name);
  }

  /**
   * 查看作业详情
   *
   * @param jobId
   * @return
   * @throws Exception
   */
  public JobDetailDto getJobDetail(String jobId) throws Exception {
    JobDTO jobDTO = this.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));
    BdpLogUtil.log4Query(jobDTO);

    JobDetailDto jobDetailDto = new JobDetailDto();
    jobDetailDto.setId(jobDTO.getId());
    jobDetailDto.setName(jobDTO.getName());
    jobDetailDto.setSrcDatasourceId(jobDTO.getSrcDatasourceId());
    jobDetailDto.setDstDatasourceId(jobDTO.getDstDatasourceId());

    DataSourceDTO srcDataSourceDTO = null;
    if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
      srcDataSourceDTO =
          dataSourceService.queryDataSourceById(
              jobDTO.getSrcDatasourceId(), jobDTO.getExtraParams());
    } else {
      srcDataSourceDTO = dataSourceService.queryDataSourceById(jobDTO.getSrcDatasourceId());
    }
    if (srcDataSourceDTO != null) {
      jobDetailDto.setSrcDatasourceName(srcDataSourceDTO.getDatasourceName());
      jobDetailDto.setSrcDatasourceType(srcDataSourceDTO.getDatasourceType());
    }

    DataSourceDTO dstDataSourceDTO =
        dataSourceService.queryDataSourceById(jobDTO.getDstDatasourceId());
    if (dstDataSourceDTO != null) {
      jobDetailDto.setDstDatasourceName(dstDataSourceDTO.getDatasourceName());
      jobDetailDto.setDstDatasourceType(dstDataSourceDTO.getDatasourceType());
    }

    String transferType = jobDTO.getTransferType();
    if (TransferType.TABLE.getDescp().equals(transferType)) {
      jobDetailDto.setTransferType(TransferType.TABLE.getRemark());
    } else if (TransferType.DATA.getDescp().equals(transferType)) {
      jobDetailDto.setTransferType(TransferType.DATA.getRemark());
    } else if (TransferType.ALL.getDescp().equals(transferType)) {
      jobDetailDto.setTransferType(TransferType.ALL.getRemark());
    } else {
      jobDetailDto.setTransferType("");
    }

    jobDetailDto.setProcessMode(ProcessMode.getDescByCode(jobDTO.getProcessMode()));

    List<TableDto> selectedTableList = new ArrayList<>();
    if (jobDTO.getJobType() == JobType.STREAM.getCode()) {
      List<TableMappingDto> tableMappingDtos =
          JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);
      if (CollectionUtils.isNotEmpty(tableMappingDtos)) {
        for (TableMappingDto tableMappingDto : tableMappingDtos) {
          TableDto tableDto = new TableDto();
          tableDto.setId(tableMappingDto.getSrc_table());
          MsgTemplateDto msgTemplateDto =
              msgTemplateService.selectByPrimaryKey(tableMappingDto.getSrc_table());
          tableDto.setSrcTableName(msgTemplateDto.getDeviceName());
          tableDto.setDstTableName(tableMappingDto.getDst_table());
          selectedTableList.add(tableDto);
        }
      }
    } else {
      TaskDTO taskDTOQuery = new TaskDTO();
      taskDTOQuery.setJobId(jobId);
      taskDTOQuery.setFlag(Enable.ZERO.getCode());
      List<TaskDTO> taskList = taskService.selectList(taskDTOQuery);
      if (CollectionUtils.isNotEmpty(taskList)) {
        for (TaskDTO task : taskList) {
          TableDto tableDto = new TableDto();
          tableDto.setId(task.getId());
          tableDto.setSrcTableName(task.getSrcTableName());
          tableDto.setDstTableName(task.getDstTableName());

          if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
            FileTemplateDTO templateDTO =
                fileTemplateService.selectByPrimaryKey(task.getSrcTableName());
            CheckParaUtils.checkDbObjectValidity(
                templateDTO,
                logger,
                String.format("没有查到数据，请确保templateId：{}的模板是否存在", task.getSrcTableName()));

            tableDto.setId(templateDTO.getId());
            tableDto.setSrcTableName(
                getSyncFilePath(task.getSrcTableName(), jobDTO.getSelectedFile()));
          }
          selectedTableList.add(tableDto);
        }
      }
    }

    Collections.sort(selectedTableList);
    jobDetailDto.setSelectedTableList(selectedTableList);
    jobDetailDto.setSelectedFile(jobDTO.getSelectedFile());

    String kafkaJson = jobDTO.getKafkaJson();
    if (StringUtils.isNotEmpty(kafkaJson)) {
      KafkaJsonDto kafkaJsonDto = JSONObject.parseObject(kafkaJson, KafkaJsonDto.class);
      jobDetailDto.setKafkaJsonDto(kafkaJsonDto);
    }
    return jobDetailDto;
  }

  public ProcessInstanceDto getProcessInstanceByProcessDefinitionId(String processDefinitionId) {
    return jobDao.getProcessInstanceByProcessDefinitionId(processDefinitionId);
  }

  public int countProcessInstanceByProcessDefinitionId(String processDefinitionId) {
    return jobDao.countProcessInstanceByProcessDefinitionId(processDefinitionId);
  }

  public SimpleJobDto getJobStatus(String jobId) throws Exception {
    JobDTO jobDTO = this.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    SimpleJobDto job = new SimpleJobDto();
    job.setJobId(jobDTO.getId());
    job.setJobStatus(jobDTO.getJobStatus());
    return job;
  }

  public TaskDTO selectByJobIdAndSrcTableName(String jobId, String srcTableName) {
    return jobDao.selectByJobIdAndSrcTableName(jobId, srcTableName);
  }

  public ProcessInstanceDto getProcessInstanceById(String processInstanceId) {
    return jobDao.getProcessInstanceById(processInstanceId);
  }

  public List<TaskInstanceDto> getSparkTaskInstanceListByProcessInstanceId(
      String processInstanceId) {
    return jobDao.getSparkTaskInstanceListByProcessInstanceId(processInstanceId);
  }

  /**
   * 调试API接口
   *
   * @param jobDTO job
   * @return 接口请求结果
   */
  public JSONObject debugApi(JobDTO jobDTO) {
    JobDTO oldJobDTO = this.selectByPrimaryKey(jobDTO.getId());
    CheckParaUtils.checkDbObjectValidity(
        oldJobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobDTO.getId()));

    APIHttpDefinition apiHttpDefinition = apiDefinitionService.getAPIHttpDefinition(jobDTO.getId());

    DataSourceDTO dataSourceDTO =
        dataSourceService.queryDataSourceById(oldJobDTO.getSrcDatasourceId());
    if (dataSourceDTO != null && dataSourceDTO.getDatasourceType().equals(Constants.DB_TYPE_API)) {
      BaseDataSource baseDataSource =
          DataSourceFactory.getDatasource(
              Constants.DB_TYPE_API, dataSourceDTO.getConnectionParams());
      ApiDataSource apiDataSource = (ApiDataSource) baseDataSource;
      apiHttpDefinition.setUrl(
          apiDataSource.getUrl() + Constants.SINGLE_SLASH + apiHttpDefinition.getUrl());
      apiHttpDefinition.setAuthType(apiDataSource.getAuthType());
      APIHttpAuth apiHttpAuth = new APIHttpAuth();
      BeanUtil.copyProperties(apiDataSource, apiHttpAuth);
      apiHttpDefinition.setAuthParam(apiHttpAuth);
    }

    Configuration readerSliceConfig = Configuration.from(JSONUtils.toJsonString(apiHttpDefinition));
    HttpUtil httpUtil = new HttpUtil(readerSliceConfig);
    String s = httpUtil.remoteCallApi();

    return JSONObject.parseObject(s);
  }

  /**
   * 重置增量信息
   *
   * @param jobId
   * @return
   * @throws Exception
   */
  public Void resetIncrement(String jobId) throws Exception {
    JobDTO jobDTO = this.selectByPrimaryKey(jobId);
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobId));

    ProcessDefinitionDto processDefinitionDto =
        jobDao.getProcessDefinitionByProcessDefinitionId(jobId);
    if (processDefinitionDto != null) {
      int code = taskJsonService.updateProcessDefinition(jobId);
      if (code == HttpStatus.SC_OK) {
        logger.info("重置增量信息成功");
      }
    }

    return null;
  }

  public String getSyncFilePath(String templateId, String selectedFile) {
    List<FileFilterRuleDto> fileFilterRuleList =
        JSONUtils.toList(selectedFile, FileFilterRuleDto.class);
    FileFilterRuleDto fileFilter = null;
    if (CollectionUtils.isNotEmpty(fileFilterRuleList)) {
      for (FileFilterRuleDto fileFilterRule : fileFilterRuleList) {
        if (fileFilterRule.getTemplateId().equals(templateId)) {
          fileFilter = fileFilterRule;
          break;
        }
      }
    }

    List<String> fileList = new ArrayList<>();
    if (fileFilter != null) {
      List<CatalogBean> catalogBeanList = fileFilter.getCatalogBeanList();
      if (CollectionUtils.isNotEmpty(catalogBeanList)) {
        fileList =
            catalogBeanList.stream()
                .map(catalogBean -> catalogBean.getCatalogName())
                .collect(Collectors.toList());
      }
    }

    return String.join(Constants.COMMA, fileList);
  }

  public class SendMsgTask implements Runnable {
    // 这里实现runnable接口
    private String infoAction;
    private JobDTO jobDTO;

    public SendMsgTask(String infoAction, JobDTO jobDTO) {
      this.infoAction = infoAction;
      this.jobDTO = jobDTO;
    }

    // 把获取的数据进行处理
    @Override
    public void run() {
      try {
        syncSendMsgToKafka(infoAction, jobDTO);
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
  }
}
